package com.wjw.redis.stream.config;

import com.wjw.redis.stream.constant.CacheConstant;
import com.wjw.redis.stream.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

/**
 * redis stream配置
 */
@Slf4j
@Configuration
public class RedisStreamConfig {
    @Autowired
    private RedisUtil redisUtil;
    @Value("${myapp.streamKey}")
    private String streamKey;

    @Value("${myapp.groupId}")
    private String groupId;

    @Value("${myapp.consumerId}")
    private String consumerId;

    @Value("${subscription:group-a}")
    private String subscriptionGroup;

    @Value("${subscription:consumer-a}")
    private String subscriptionConsumer;

    @Value("${redis.stream.key:redis_steam_01}")
    private String redisStreamKey;
    private final static String[] GROUPS = {"group-a", "group-b"};
    private final static String[] STREAMS = {CacheConstant.REDIS_STEAM_01, CacheConstant.REDIS_STEAM_02};
    private static final Lock initLock = new ReentrantLock();
    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * 初始化stream和group。该方法在对象创建后立即调用，用于确保启动时不会因为stream或group未初始化而报异常。
     */
    @PostConstruct
    public void init() {
        try {
            initLock.lock(); // 获取初始化锁，确保线程安全
            boolean hasKey = redisUtil.hasKey(streamKey); // 检查streamKey是否已存在于Redis中
            if (!hasKey) {
                // 如果streamKey不存在，则初始化stream信息
                HashMap<String, String> streamInfoMap = new HashMap<>();
                streamInfoMap.put("id", streamKey);
                streamInfoMap.put("name", "CacheConstant");
                redisUtil.addMap(streamKey, streamInfoMap);
            }
            // 初始化group
            initializeGroups();
        } finally {
            initLock.unlock(); // 释放初始化锁
        }
    }

    /**
     * 初始化指定的流组。该方法会查询当前流中已经存在的组，然后对比预定义的组名列表，
     * 对于在流中不存在的组，将其在流中创建。
     */
    private void initializeGroups() {
        try {
            // 从Redis中查询当前流的所有组信息
            StreamInfo.XInfoGroups groups = redisUtil.queryGroups(streamKey);
            // 将查询到的组名转换为Set集合，以便进行后续的比较
            Set<String> groupNames = groups.stream().map(StreamInfo.XInfoGroup::groupName).collect(Collectors.toSet());
            // 遍历预定义的组名列表
            for (String group : GROUPS) {
                // 如果当前组名不在查询到的组名集合中，则创建该组
                if (!groupNames.contains(group)) {
                    redisUtil.createGroup(streamKey, group);
                }
            }
        } catch (Exception e) {
            // 记录初始化流组失败的错误日志
            log.error("Failed to initialize stream groups", e);
        }
    }

    @Bean
    public StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ?> streamMessageListenerContainerOption() {

        AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(), r -> {
            Thread thread = new Thread(r);
            thread.setName("async-stream-consumer-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        });

        return StreamMessageListenerContainer.StreamMessageListenerContainerOptions
            .builder()
            // 一次最多获取多少条消息
            .batchSize(5)
            // 运行 Stream 的 poll task
            .executor(executor)
            // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小
            .pollTimeout(Duration.ofSeconds(3))
            // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
            .errorHandler(new StreamErrorHandler())
            .build();
    }

    @Bean
    public StreamMessageListenerContainer<String, ?> streamMessageListenerContainer(RedisConnectionFactory factory) {
        StreamMessageListenerContainer<String, ?> listenerContainer = StreamMessageListenerContainer.create(factory, streamMessageListenerContainerOption());
        listenerContainer.start();
        return listenerContainer;
    }

    /**
     * 创建一个订阅对象。这个方法尝试使用给定的流消息监听器容器来创建一个订阅。
     * 如果创建过程中遇到异常，会记录错误日志，并根据实际情况处理异常，可能返回null。
     *
     * @param streamMessageListenerContainer 流消息监听器容器，用于监听和处理消息。
     * @return 返回创建的订阅对象，如果创建失败则返回null。
     */
    @Bean
    public Subscription subscription(StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer) {
        try {
            // 确保参数使用配置化的值
            return createSubscription(streamMessageListenerContainer);
        } catch (Exception e) {
            log.error("Failed to create subscription", e);
            // 根据实际情况处理异常，例如返回null或抛出自定义异常
            return null;
        }
    }

    /**
     * 创建一个订阅，用于监听流消息。
     *
     * @param streamMessageListenerContainer 流消息监听器容器，负责处理接收到的消息。
     * @return 返回创建的订阅对象。
     */
    private Subscription createSubscription(StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer) {
        // 使用指定的消费者信息、流偏移量和消费监听器创建订阅
        return streamMessageListenerContainer.receiveAutoAck(Consumer.from(groupId, consumerId),
            StreamOffset.create(streamKey, ReadOffset.lastConsumed()), new ConsumeListener(redisUtil));
    }


    @Bean
    public Subscription subscription1(StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer) {
        try {
            return streamMessageListenerContainer.receive(Consumer.from(subscriptionGroup, subscriptionConsumer),
                StreamOffset.create(streamKey, ReadOffset.lastConsumed()), new ConsumeListener1());
        } catch (Exception e) {
            // 适当的异常处理逻辑
            log.error("Failed to create subscription", e);
            // 根据情况，可以选择返回null或者抛出一个自定义的异常
            return null;
        }
    }

}
